在完成地端推論環境部署(Day 03)與多雲端 Provider 抽象層(Day 04)後,今天我們要動手打造中台的統一入口:API Gateway 核心骨架。
Gateway 是整個 AI 中台的神經樞紐,所有來自內部開發者、IDE 外掛或自動化腳本的請求都必須先經過此處。今天我們將使用高效能的 FastAPI,建立一個相容於 OpenAI 規格的 /v1/chat/completions 端點,並實作 API Key 權限驗證與全域例外處理機制。
FallbackLLMManager 執行。ai-gateway/
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI 核心入口與端點
│ ├── config.py # 設定檔與 Key 管理
│ ├── auth.py # API Key 驗證依賴項
│ ├── models.py # Pydantic 資料模型 (Day 04 定義)
│ ├── providers.py # 多模型 Client (Day 04 定義)
│ └── manager.py # Fallback 管理器 (Day 04 定義)
├── requirements.txt
└── test_gateway.py # 端點測試腳本
fastapi>=0.110.0
uvicorn>=0.29.0
pydantic>=2.6.0
python-dotenv>=1.0.0
在生產環境中,API Key 通常儲存於資料庫或快取。此處先以環境變數或記憶體字典模擬企業各部門的 Key 配置:
import os
from typing import Dict
# 模擬企業內部合法 Key 及其所屬部門
VALID_API_KEYS: Dict[str, str] = {
os.getenv("GATEWAY_KEY_RD", "sk-corp-rd-team-1001"): "Research & Development",
os.getenv("GATEWAY_KEY_QA", "sk-corp-qa-team-2002"): "Quality Assurance"
}
使用 FastAPI 的 Security 與 HTTPBearer 抽取請求中的 Bearer Token:
from fastapi import Security, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from app.config import VALID_API_KEYS
security = HTTPBearer()
def verify_api_key(credentials: HTTPAuthorizationCredentials = Security(security)) -> str:
token = credentials.credentials
if token not in VALID_API_KEYS:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="無效的 API Key 或未經授權的訪問請求",
headers={"WWW-Authenticate": "Bearer"},
)
# 回傳該 Key 所屬部門名稱供 Context 使用
return VALID_API_KEYS[token]
建立 /v1/chat/completions 端點,注入鑑權依賴並串接 Provider 管理器:
from fastapi import FastAPI, Depends, HTTPException, Request
from fastapi.responses import JSONResponse
import logging
from app.models import ChatCompletionRequest, ChatCompletionResponse
from app.auth import verify_api_key
from app.providers import OpenAIProvider, ClaudeProvider, GeminiProvider
from app.manager import FallbackLLMManager
# 初始化日誌
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("AIGateway")
app = FastAPI(
title="Enterprise AI Gateway",
version="1.0.0",
description="企業 AI 中台統一核心網關"
)
# 初始化多模型降級管理器
llm_manager = FallbackLLMManager([
OpenAIProvider(),
ClaudeProvider(),
GeminiProvider()
])
# 全域例外處理器
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
logger.error(f"Gateway 處理未預期錯誤: {str(exc)}")
return JSONResponse(
status_code=500,
content={
"error": {
"message": "中台服務內部錯誤,請聯繫系統管理員",
"type": "internal_server_error",
"detail": str(exc)
}
}
)
# 健康檢查端點
@app.get("/health")
async def health_check():
return {"status": "ok", "service": "Enterprise AI Gateway"}
# 核心端點:Chat Completions
@app.post(
"/v1/chat/completions",
response_model=ChatCompletionResponse,
summary="統一對話生成接口"
)
async def create_chat_completion(
request: ChatCompletionRequest,
department: str = Depends(verify_api_key)
):
logger.info(f"收到來自 [{department}] 的生成請求,模型指向: {request.model}")
try:
# 交由 Fallback 管理器執行調用
response = llm_manager.execute_with_fallback(request)
return response
except Exception as e:
logger.error(f"請求處理失敗: {str(e)}")
raise HTTPException(status_code=502, detail=f"上游模型服務調用失敗: {str(e)}")
uvicorn app.main:app --host 0.0.0.0 --port 8080 --reload
curl -X POST http://localhost:8080/v1/chat/completions \
-H "Content-Type: application/json" \
-d '{
"messages": [{"role": "user", "content": "Hello"}]
}'
回應:
{"detail": "Not authenticated"}
curl -X POST http://localhost:8080/v1/chat/completions \
-H "Authorization: Bearer sk-corp-rd-team-1001" \
-H "Content-Type: application/json" \
-d '{
"messages": [
{"role": "user", "content": "請列出三個 Clean Code 的核心原則"}
],
"temperature": 0.2
}'
回應:
{
"provider": "openai",
"model": "gpt-4o-mini",
"content": "Clean Code 的三個核心原則包括:1. 單一職責原則... 2. 有意義的命名... 3. 避免重複 (DRY)...",
"usage": {
"prompt_tokens": 18,
"completion_tokens": 120,
"total_tokens": 138
}
}
中台 Gateway 基礎框架已經建立。明天 Day 06 我們將進入向量存儲的核心環節:向量資料庫選型與建置,實際搭建本地 Qdrant 向量資料庫,並規劃企業規範知識庫的 Collections 與 Payload 結構。